/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package it.txt.access.capability.libraries.rabbitmq.impl.subscriber;

import com.rabbitmq.client.LongString;
import com.rabbitmq.client.QueueingConsumer;
import it.txt.access.capability.libraries.rabbitmq.AbstractApplication;
import it.txt.access.capability.libraries.rabbitmq.exception.AbstractApplicationException;
import it.txt.access.capability.libraries.rabbitmq.impl.subscriber.listener.SubscriberApplicationListener;
import it.txt.access.capability.libraries.rabbitmq.impl.subscriber.listener.SubscriberMessageModel;
import it.txt.access.capability.libraries.rabbitmq.settings.AbstractApplicationSettings;
import static it.txt.access.capability.libraries.rabbitmq.impl.publisher.PublisherApplication.*;
import java.io.IOException;
import java.net.URI;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Level;

/**
 *
 * @author seccia
 */
public class SubscriberApplication extends AbstractApplication implements Runnable {

    private static HashMap<String, SubscriberApplicationListener> listeners =
            new HashMap<String, SubscriberApplicationListener>();
    private volatile Thread subscriberThread;
    private QueueingConsumer rabbitConsumer;
    private boolean isSubscribed;

    public SubscriberApplication(AbstractApplicationSettings settings) throws AbstractApplicationException {
        super(settings);
    }

    public static void addSubscriberMessagesListener(String listenerID, 
            SubscriberApplicationListener listener) throws AbstractApplicationException{
        if(listeners.containsKey(listenerID)){
            throw new AbstractApplicationException("Cannot add message listener "
                    + "becouse ID: " + listenerID + " is already been used.");
        }
        listeners.put(listenerID, listener);
    }

    public static void removeSubscriberMessagesListener(String listenerID) throws AbstractApplicationException{
        if(!listeners.containsKey(listenerID)){
            throw new AbstractApplicationException("Cannot find message listener with ID: " + listenerID);
        }
        listeners.remove(listenerID);
    }

    /**
     * The authorization server has created the queue for the subscriber by
     * using the routing key specified into the request.
     */
    public void basicSubscribe() throws AbstractApplicationException {
        try {
            if (isSubscribed()) {
                throw new AbstractApplicationException("Application already subscribed.");
            }
            boolean autoAck = false;
            LOGGER.log(Level.INFO, "Subscribing on: {0}", getCapabilityConnection().getQueueName());
            rabbitConsumer = new QueueingConsumer(getCapabilityConnection().getChannel());
            getCapabilityConnection().getChannel().basicConsume(getCapabilityConnection().getQueueName(),
                    autoAck, rabbitConsumer);
            subscriberThread = new Thread(this);
            subscriberThread.start();
            isSubscribed = true;
        } catch (IOException ex) {
            throw new AbstractApplicationException(ex.getMessage(), ex);
        }
    }

    public void run() {
        try {
            Thread thisThread = Thread.currentThread();
            LOGGER.info("Waiting for 'Messages'...");
            while (subscriberThread == thisThread) {
                QueueingConsumer.Delivery delivery = rabbitConsumer.nextDelivery();
                onMessage(delivery);
                getCapabilityConnection().getChannel().basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
        catch (Exception ex) {
            if (!isSubscribed()) {
                LOGGER.info("Subscriber stopped");
                return;
            }
            LOGGER.log(Level.SEVERE, ex.getMessage(), ex);
            throw new RuntimeException(ex.getMessage(), ex);
        }
    }

    private void onMessage(QueueingConsumer.Delivery delivery) throws IOException {
        LOGGER.info("Message Received...");
        LOGGER.log(Level.INFO, "Content: {0}", new String(delivery.getBody()));

        Map<String, Object> headers = delivery.getProperties().getHeaders();

        try {
            LongString publisherIDBytes = (LongString) headers.get(CHANNEL_PUB_PROPS_ID);
            String publisherID = new String(publisherIDBytes.getBytes());
            LOGGER.log(Level.INFO, "Publisher ID: {0}", publisherID);

            long timeStampLong = (Long) headers.get(CHANNEL_PUB_PROPS_TIME_STAMP_LONG);
            LOGGER.log(Level.INFO, "Timestamp: {0}", new Date(timeStampLong));

            LongString entityURIBytes = (LongString) headers.get(CHANNEL_PUB_PROPS_ENTITY_URI);
            String entityURI = new String(entityURIBytes.getBytes());
            LOGGER.log(Level.INFO, "Publisher Entity URI: {0}", entityURI);
            
            LongString entityDataURIBytes = (LongString) headers.get(CHANNEL_PUB_PROPS_ENTITY_DATA_URI);
            String entityDataURI = new String(entityDataURIBytes.getBytes());
            LOGGER.log(Level.INFO, "Publisher Entity Data URI: {0}", entityDataURI);
                        
            SubscriberMessageModel messageModel = new SubscriberMessageModel();
            messageModel.setMessageTimeStamp(new Date(timeStampLong));
            messageModel.setEntityDataURI(new URI(entityDataURI));
            messageModel.setEntityURI(new URI(entityURI));
            messageModel.setPublisherSubjectID(publisherID);

            for(SubscriberApplicationListener listener : listeners.values()){
                listener.onSubscriberMessagege(messageModel);
            }
                        
        } catch (Exception e) {
            e.printStackTrace();
            LOGGER.log(Level.SEVERE, "{0} : Some Required have not been found.", e.getMessage());
        }        
    }

    
    @Override
    public void sessionRenegotiation() throws AbstractApplicationException {
        throw new UnsupportedOperationException("Not supported yet.");
    }

    @Override
    public void disconnect() throws AbstractApplicationException{
        isSubscribed = false;
        super.disconnect();
    }

    public synchronized boolean isSubscribed() {
        return isSubscribed;
    }
}
